package com.hexinfo.dmpro.monitor.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.monitor.model.YarnAppModel;
import com.hexinfo.dmpro.monitor.model.base.MonitorReq;
import com.hexinfo.dmpro.monitor.model.base.MonitorRsp;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author peng
 * @date 2022/2/15 15:18
 */
@Slf4j
@Service
public class CmYarnAppService {

    private int SUCESS_CODE = 200;

    @Autowired
    private CmApiService cmApiService;

    private String apiUrl = "/clusters/{clusterName}/services/{serviceName}/yarnApplications?limit=1000&from={startTime}";

    public List<YarnAppModel> getYarnApps(MonitorReq req) {

        List<YarnAppModel> yarnAppModels = new ArrayList<>();

        JSONObject jsonObject = JSONObject.parseObject(req.getOtherParam());
        //配置cm Api的 url
        String url = jsonObject.get("cmApi").toString();
        req.setUrl(url);
        String user = jsonObject.get("user").toString();
        String password = jsonObject.get("password").toString();

        Map header = new HashMap<>();
        try {
            header.put("Authorization", "Basic " + Base64.getEncoder().encodeToString((user + ":" + password).getBytes("utf-8")));
            req.setHeader(header);
        } catch (UnsupportedEncodingException e) {
            log.error("设置Authorization异常", e);
        }
        JSONArray jsonArray = null;
        url += apiUrl;
        String startTime = req.getStartTime();
        startTime = getISO86Time(startTime);
        req.setUrl(url.replace("{clusterName}", req.getClusterFlag()).replace("{startTime}", startTime).replace("{serviceName}",req.getServiceName()));
        MonitorRsp rsp = cmApiService.execute(req);
        if (SUCESS_CODE == rsp.getCode()) {
            String body = rsp.getBody();
            JSONObject obj = JSONObject.parseObject(body);
            jsonArray = obj.getJSONArray("applications");
            if (jsonArray != null && jsonArray.size() > 0) {
                for (int i = 0; i < jsonArray.size(); i++) {
                    JSONObject objApp = jsonArray.getJSONObject(i);

                    YarnAppModel yarnAppModel = new YarnAppModel();
                    yarnAppModel.setId(objApp.getString("applicationId").replace("job_", "application_"));
                    yarnAppModel.setUser(objApp.getString("user"));
                    yarnAppModel.setName(objApp.getString("name"));
                    yarnAppModel.setQueue(objApp.getString("pool"));
                    yarnAppModel.setState(objApp.getString("state"));
                    yarnAppModel.setFinalStatus(objApp.getString("state"));
                    yarnAppModel.setStartedTime(getBjTime(objApp.getString("startTime")));
                    yarnAppModel.setApplicationType("MAPREDUCE");

                    if (objApp.get("endTime") != null) {
                        yarnAppModel.setFinishedTime(getBjTime(objApp.getString("endTime")));
                    }
                    if (objApp.get("attributes") != null) {
                        JSONObject attributes = objApp.getJSONObject("attributes");
                        if (attributes.get("mb_millis") != null) {
                            yarnAppModel.setMbMillis(attributes.get("mb_millis").toString());
                        }
                        if (attributes.get("hdfs_bytes_written") != null) {
                            yarnAppModel.setHdfsBytesWritten(attributes.get("hdfs_bytes_written").toString());
                        }
                        if (attributes.get("hdfs_bytes_read") != null) {
                            yarnAppModel.setHdfsBytesRead(attributes.get("hdfs_bytes_read").toString());
                        }
                        if (attributes.get("file_bytes_written") != null) {
                            yarnAppModel.setFileBytesWritten(attributes.get("file_bytes_written").toString());
                        }
                        if (attributes.get("file_bytes_read") != null) {
                            yarnAppModel.setFileBytesRead(attributes.get("file_bytes_read").toString());
                        }
                        yarnAppModels.add(yarnAppModel);
                    }
                }
            }
        }
        return yarnAppModels;
    }

    /**
     * 运行作业信息，只获取cm上的数据
     * @return
     */
    public List<YarnAppModel> getYarnApps1(MonitorReq req) {

        List<YarnAppModel> yarnAppModels = new ArrayList<>();

        JSONObject jsonObject = JSONObject.parseObject(req.getOtherParam());
        //配置cm Api的 url
        String url = jsonObject.get("cmApi").toString();
        req.setUrl(url);
        String user = jsonObject.get("user").toString();
        String password = jsonObject.get("password").toString();

        Map header = new HashMap<>();
        try {
            header.put("Authorization", "Basic " + Base64.getEncoder().encodeToString((user + ":" + password).getBytes("utf-8")));
            req.setHeader(header);
        } catch (UnsupportedEncodingException e) {
            log.error("设置Authorization异常", e);
        }
        JSONArray jsonArray = null;
        url += apiUrl;
        String startTime = req.getStartTime();
        startTime = getISO86Time(startTime);
        req.setUrl(url.replace("{clusterName}", req.getClusterFlag()).replace("{startTime}", startTime).replace("{serviceName}",req.getServiceName()));
        MonitorRsp rsp = cmApiService.execute(req);
        if (SUCESS_CODE == rsp.getCode()) {
            String body = rsp.getBody();
            JSONObject obj = JSONObject.parseObject(body);
            jsonArray = obj.getJSONArray("applications");
            if (jsonArray != null && jsonArray.size() > 0) {
                for (int i = 0; i < jsonArray.size(); i++) {
                    JSONObject objApp = jsonArray.getJSONObject(i);

                    YarnAppModel yarnAppModel = new YarnAppModel();
                    yarnAppModel.setId(objApp.getString("applicationId").replace("job_", "application_"));
                    yarnAppModel.setUser(objApp.getString("user"));
                    yarnAppModel.setName(objApp.getString("name"));
                    yarnAppModel.setQueue(objApp.getString("pool"));
                    yarnAppModel.setState(objApp.getString("state"));
                    yarnAppModel.setFinalStatus(objApp.getString("state"));
                    yarnAppModel.setStartedTime(getBjTime(objApp.getString("startTime")));
                    yarnAppModel.setApplicationType("MAPREDUCE");

                    if (objApp.get("endTime") != null) {
                        yarnAppModel.setFinishedTime(getBjTime(objApp.getString("endTime")));
                    }
                    if (objApp.get("attributes") != null) {
                        JSONObject attributes = objApp.getJSONObject("attributes");
                        if (attributes.get("mb_millis") != null) {
                            yarnAppModel.setMbMillis(attributes.get("mb_millis").toString());
                        }
                        if (attributes.get("vcores_millis") != null) {
                            yarnAppModel.setVcoresMillis(attributes.get("vcores_millis").toString());
                        }
                        if (attributes.get("hdfs_bytes_written") != null) {
                            yarnAppModel.setHdfsBytesWritten(attributes.get("hdfs_bytes_written").toString());
                        }
                        if (attributes.get("hdfs_bytes_read") != null) {
                            yarnAppModel.setHdfsBytesRead(attributes.get("hdfs_bytes_read").toString());
                        }
                        if (attributes.get("file_bytes_written") != null) {
                            yarnAppModel.setFileBytesWritten(attributes.get("file_bytes_written").toString());
                        }
                        if (attributes.get("file_bytes_read") != null) {
                            yarnAppModel.setFileBytesRead(attributes.get("file_bytes_read").toString());
                        }
                        yarnAppModels.add(yarnAppModel);
                    }
                }
            }
        }
        return yarnAppModels;
    }

    public static String getISO86Time(String date) {
        if (StringUtils.isBlank(date)) {
            return "";
        }
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date newTime = null;
        try {
            newTime = format.parse(date);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        TimeZone tz = TimeZone.getTimeZone("GMT-01");
        DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
        df.setTimeZone(tz);
        return df.format(newTime);
    }

    public static String getBjTime(String ISOdate) {
        if (StringUtils.isBlank(ISOdate)) {
            return "";
        }
        DateTimeFormatter dtf1 = DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
        DateTime dt = dtf1.parseDateTime(ISOdate);
        DateTimeFormatter dtf2 = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
        return dt.toString(dtf2);
    }

}
